#   导入需要用到的包
import pandas as pd
from hdfs import InsecureClient             #   用于查看hdfs文件内容的包
from io import StringIO                     #   用于将字符串模拟成文件对象
import pymysql
from hdfs.client import Client

#   对dataframe的数据展示进行设置，这样让观看数据更加方便舒服
#   显示所有列
pd.set_option('display.max_columns',None)
#   显示所有行
pd.set_option('display.max_rows',None)
#   设置value的显示长度
pd.set_option('max_colwidth',100)
#   设置1000列时才换行
pd.set_option('display.width',1000)


#   创建连接hdfs(9870是hadoop的web端口号)
hdfs_connect=InsecureClient(url='http://192.168.40.110:9870',user='root')

#   hdfs_connect.read('/llp/all_datas.csv') as file：将数据读取到并且赋值给file
#   file.data表示file里面有一个属性，属性的值就是文本的内容
#   decode('utf-8'):对二进制的文件解码为utf-8类型的
with hdfs_connect.read('/llp/all_datas.csv') as file:
    datas=file.data.decode('utf-8')

#   将读取到的字符串模拟成文件对象，这样才可以使用read_csv函数读取成dataframe
str_to_file=StringIO(datas)
#   将数据读取成dataframe
data=pd.read_csv(str_to_file,sep=',',names=['name','position','money','corporation_name','trait','demand','url'])
# print(data.columns)


#   1.检查是否有重复数据，有的话删除，没有的话忽略
#   删除重复数据前的数据长度
# print("删除重复数据前的数据数量:",len(data))
#   any()是检查所有行，是否有一个true
if data.duplicated().any():
    data.drop_duplicates(inplace=True)
else:
    print("没有重复数据")
#   删除重复数据后的数据长度
# print("删除重复数据后的数据数量:",len(data))

print(data['money'])
#   2.对薪资待遇一列数据进行规整化(将其全部换算成每个月多少钱)
money_datas=data['money']
#   2.1首先对没有填写工资待遇记为0
money_have_zero=[]
for x in money_datas:
    if x=='没':
        money_have_zero.append(0)
    else:
        money_have_zero.append(x)
#   将data的薪资一列更改，没有填写薪资待遇的记为0：
data['money']=money_have_zero
# print(data['money'])

#   2.2处理120元/天这种类型，将薪资待遇为多少钱每天的换算为一个月按30天来算，是多少钱
#   2.2.1把薪资里面的14薪这样的带多少薪的都去掉()
data['money']=data['money'].apply(lambda x: str(x).split('·')[0] if '薪'  in str(x) else str(x))
# print(data['money'])

#   2.2.2首先把/天去掉
data['money']=data['money'].apply(lambda x: str(x).strip('/天') if "/天" in str(x) else str(x) )
# print(data['money'])

#   2.2.3将多少元换算成120*30的结果
data['money']=data['money'].apply(lambda x: int(str(x).split('元')[0])*30 if "元" in str(x) else str(x) )
# print(data['money'])

#   2.3 处理10-20万/年这种数据类型的，将其除以12，并且保留最大值和最小值
data['money']=data['money'].apply(
    lambda x:
    [int((float(str(x).strip('万/年').split('-')[0])*10000)/12),int((float(str(x).strip('万/年').split('-')[1])*10000)/12)]
    if "万/年" in str(x)
    else str(x)
)
# print(data['money'])


#   2.4处理3千及以下这样的数据类型
data['money']=data['money'].apply(lambda x: int(float(str(x).strip('千及以下'))*1000) if "千及以下" in str(x) else str(x) )

# #   2.5处理8千-1万这样的类型
data['money']=data['money'].apply(
    lambda x:
    [int(float(str(x).replace("万","").replace("千","").split("-")[0])*1000),int(float(str(x).replace("万","").replace("千","").split("-")[1])*10000)]
    if "万" in str(x) and "千" in str(x)
    else str(x)
)
print(data['money'])

#   2.6处理1.3-1.5万这样的类型的
data['money']=data['money'].apply(
    lambda x:
    [int(float(str(x).strip('万').split('-')[0])*10000),int(float(str(x).strip('万').split('-')[1])*10000)]
    if "万"  in str(x)
    else str(x)
)
# print(data['money'])

#   2.7处理6-8千这样的数据类型
data['money']=data['money'].apply(
    lambda x:
    [int(float(str(x).strip('千').split('-')[0])*1000),int(float(str(x).strip('千').split('-')[1])*1000)]
    if "千" in str(x)
    else str(x)
)

# print(data['position'])
#   3.处理地区,将工作所在地的城市保留，所在地区去掉
data['position']=data['position'].apply(lambda x: str(x).split('-')[0] if "-" in str(x) else str(x) )
print(data['position'])

# print("新增列前的字段：",data.columns)
#   4.新增两列数据，一个存最小的薪资，一个存最大的薪资，如果只有一个的话，就用那一个代替
#   定义列表存储最小和最大的薪资
min_list=[]
max_list=[]
for x in data['money']:
    str_data=str(x)
    if ',' in str_data:
        data_list=str_data.replace("[","").replace("]","").split(",")
        min_list.append(data_list[0])
        max_list.append(data_list[1])
    else:
        min_list.append(x)
        max_list.append(x)
data['max_money']=max_list
data['min_money']=min_list
# print("新增列后的字段：",data.columns)


#   5.将薪资待遇存在最大值和最小值的数据进行取平均值
data['money']=data['money'].apply(
    lambda x:
    int((int(str(x).strip("[").strip("]").split(",")[0])+int(str(x).strip("[").strip("]").split(",")[1]))/2)
    if ',' in str(x)
    else str(x)
)
print(data['money'])

#   6.将没有标注薪资的行给去掉()
data=data[data['money']!=str(0)]
# print(data['money'])

#   7.将其他字段(trait,demand)值的方括号去掉
data['trait']=data['trait'].apply(lambda x: str(x).replace('[','').replace(']','') if '[' in str(x) else str(x))
data['demand']=data['demand'].apply(lambda x: str(x).replace('[','').replace(']','') if '[' in str(x) else str(x))
print(data)


#   将数据写入到mysql
#   建立与mysql的连接
connect=pymysql.connect(host='192.168.40.110',user='root',password='123456',database='llp',charset='utf8')
#   通过连接创建游标对象
cursor=connect.cursor()
#   定义sql如果存在表就删除表(cleanse：清洗)
drop_table_sql="drop table if exists cleanse;"
#   执行删除语句
cursor.execute(drop_table_sql)
#   定义创建表的语句
create_table_sql="""
create table if not exists cleanse(
                    id  int primary key AUTO_INCREMENT,
                    name varchar(300),
                    position varchar(300),
                    money int,
                    corporation_name varchar(50),
                    trait varchar(300),
                    demand varchar(300),
                    url varchar(300),
                    max_money int,
                    min_money int
                    );
"""
#   执行创建表的语句
cursor.execute(create_table_sql)
#   定义将数据插入的语句
insert_sql="insert into  cleanse(name,position,money,corporation_name,trait,demand,url,max_money,min_money) values (%s,%s,%s,%s,%s,%s,%s,%s,%s);"
#   转换成[  [1,2,3],[4,5,6]     ]的格式，方便写入mysql    data.values.tolist()会将行数据转换成列表，不包括索引
data_list=data.values.tolist()
#   执行插入数据
cursor.executemany(insert_sql,data_list)

#   提交任务(只有提交任务才会执行sql)
connect.commit()
#   关闭游标对象
cursor.close()
#   关闭与mysql的连接
connect.close()

#   将数据保存到本地一份
data.to_csv('cleanse.csv',index=False,encoding='utf-8')

#   将文件上传到hdfs
#   创建一个hdfs的客户端实例(9870是web端访问hdfs的端口号)
client=Client("http://192.168.40.110:9870")

#   hdfs上面的文件路径
hdfs_path='/llp/'

#    使用客户端的upload方法将文件上传到hdfs
client.upload(hdfs_path=hdfs_path,local_path='cleanse.csv',overwrite=True)






















